package com.xiaoyi.springboot.service;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Maps;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.xiaoyi.springboot.listener.ListenerAckFailure;
import com.xiaoyi.springboot.listener.MQMessage;
import com.xiaoyi.springboot.listener.RabbitMqRerurnBack;
import com.xiaoyi.springboot.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeoutException;

@ConditionalOnProperty(
        value = {"spring.rabbitmq.enable"},
        havingValue = "true"
)
@Component
public class RabbitTemplateSetvice implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback, InitializingBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitTemplateSetvice.class);
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private Connection connection;
    @Autowired
    private RabbitMqRerurnBack returnBack;
    @Autowired
    private ListenerAckFailure listenerAckFailure;
    public RabbitTemplateSetvice() {
    }

    /**
     * 发送带有确认信息的消息
     * @param exChange
     * @param routKey
     * @param mqMessage
     * @return
     */
    public Boolean sendConfirmMsg(String exChange, String routKey, MQMessage mqMessage) {
        String content = JSON.toJSONString(mqMessage);
        Channel channel = null;

        try {
            channel = this.connection.createChannel(false);
            channel.confirmSelect();
            channel.basicPublish(exChange, routKey, (AMQP.BasicProperties)null, content.getBytes("UTF-8"));
            if (channel.waitForConfirms()) {
                channel.close();
                Boolean var6 = Boolean.TRUE;
                return var6;
            }
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        } finally {
            try {
                if (channel != null) {
                    channel.close();
                }
            } catch (TimeoutException | IOException var16) {
                LOGGER.error(var16.getMessage(), var16);
            }

        }

        return Boolean.FALSE;
    }

    /**
     * 发送消息
     * @param exChange 交换机
     * @param routKey 路由
     * @param mqMessage 消息队列
     * @return
     */
    public Boolean sendMsg(String exChange, String routKey, MQMessage mqMessage) {
        if (mqMessage == null) {
            return Boolean.FALSE;
        } else {
            String content = JSON.toJSONString(mqMessage);
            if (null != content && !"".equals(content)) {
                Map<String, Object> map = Maps.newHashMap();
                map.put("id", UUID.randomUUID());
                map.put("exChange", exChange);
                map.put("routKey", routKey);
                map.put("content", mqMessage);
                CorrelationData correlationId = new CorrelationData(JSON.toJSONString(map));

                try {
                    this.rabbitTemplate.convertAndSend(exChange, routKey, content.getBytes("UTF-8"), correlationId);
                } catch (UnsupportedEncodingException | AmqpException var8) {
                    LOGGER.error(var8.getMessage(), var8);
                    return Boolean.FALSE;
                }

                return Boolean.TRUE;
            } else {
                return Boolean.FALSE;
            }
        }
    }

    /**
     * 确认消息
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            if (correlationData == null || correlationData.getId() == null) {
                this.returnBack.returnCallback((String)null, (String)null, (String)null);
                return;
            }

            Map<String, Object> json = JSONObject.parseObject(correlationData.getId());
            this.returnBack.returnCallback(StringUtils.objToString(json.get("exChange")), StringUtils.objToString(json.get("routKey")), StringUtils.objToString(json.get("content")));
        }

    }
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        this.returnBack.returnCallback(exchange, routingKey, new String(message.getBody()));
    }
    @Override
    public void afterPropertiesSet() throws Exception {
        this.rabbitTemplate.setConfirmCallback(this::confirm);
        this.rabbitTemplate.setReturnCallback(this::returnedMessage);
    }

}

